package com.kafka

import java.time.LocalDateTime
import java.util.Properties

import kafka.producer.{KeyedMessage, Producer, ProducerConfig}

import scala.io.Source
import scala.util.control.Breaks._

object PutMysqlRecords {

  def main(args: Array[String]): Unit = {

    println("program is running.............")

    //    val fileName = "C:\\Users\\zhangbn\\Desktop\\1.txt"
    //    val source = Source.fromFile(fileName)
    //    val lines = source.getLines

    val fileName = args(1)

    val props = new Properties()
    props.put("metadata.broker.list", "10.82.27.22:9092,10.82.27.23:9092,10.82.27.24:9092") // broker 如果有多个,中间使用逗号分隔
    props.put("serializer.class", "kafka.serializer.StringEncoder")
    props.put("request.required.acks", "1")
    val config = new ProducerConfig(props)
    val producer = new Producer[String, String](config)

    val startTime: Long = System.currentTimeMillis

    val MAX: Int = Integer.valueOf(args(0))
    var count: Int = 0
    var times: Int = 0
    while (count < MAX) {
      times = times + 1
      val source = Source.fromFile(fileName)
      val lines = source.getLines
      breakable {
        for (line <- lines) {
          val msg = line
          val topic = "vehicle"
          val data = new KeyedMessage[String, String](topic, line)
          producer.send(data)
          count = count + 1
          if(count==1){
            println(line)
          }
          if (count >= MAX) {
            break
          }
        }
      }
      source.close()
      val timeConsuming: Long = System.currentTimeMillis
      println(s"MAX=${MAX}  times=${times}  count=${count}  timeConsuming=${timeConsuming-startTime}ms  now=${LocalDateTime.now()}")
    }

    producer.close()
    val endTime: Long = System.currentTimeMillis
    System.out.println("程序运行时间： " + (endTime - startTime) + "ms")
  }

}
